package com.darcytech.debezium.core.rpc;

import java.net.UnknownHostException;
import java.util.concurrent.atomic.AtomicInteger;

import com.darcytech.debezium.common.utils.NetUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.darcytech.debezium.common.utils.ThreadUtils;
import com.darcytech.debezium.core.api.impl.ApiSender;
import com.darcytech.debezium.core.pool.ConnectorConnectionBinding;

public class BootStrap {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConnectorConnectionBinding.class);

    private static final AtomicInteger threadNum = new AtomicInteger();

    static String LOCAL_HOST = "localhost";

    private final RpcServer rpcServer;

    private final RpcClient rpcClient;

    private final String rpcAddress;

    private final ConnectorConnectionBinding connectorConnectionBinding;

    private final LeaderLatch leaderLatch;

    private final LeaderLatchChecker leaderLatchChecker;

    private Thread serverThread;

    public BootStrap(final String zkServers, final String zkPath, final String rpcAddress,
                     final ConnectorConnectionBinding connectorConnectionBinding, final ApiSender apiSender) {
        String calcRpcAddress;
        if (rpcAddress == null || rpcAddress.equals("")) {
            try {
                calcRpcAddress = NetUtils.getIpAddress() + ":" + RpcServer.DEFAULT_ADDRESS;
            } catch (UnknownHostException e) {
                calcRpcAddress = LOCAL_HOST + ":" + RpcServer.DEFAULT_ADDRESS;
            }

        } else {
            calcRpcAddress = rpcAddress;
        }
        this.rpcAddress = calcRpcAddress;

        this.rpcServer = new RpcServer(connectorConnectionBinding, this.rpcAddress);

        this.connectorConnectionBinding = connectorConnectionBinding;

        CuratorFramework client = CuratorFrameworkFactory.newClient(zkServers, new ExponentialBackoffRetry(1000, 3));
        client.start();
        this.leaderLatch = new LeaderLatch(client, zkPath, this.rpcAddress);
        this.leaderLatch.addListener(new Listener());
        this.leaderLatchChecker = new LeaderLatchChecker();

        this.rpcClient = new RpcClient(leaderLatch, apiSender);
    }

    /**
     * 1.启动leader选举，当前节点选为主节点时触发{@link Listener#isLeader}，开启RpcServer，并装载ConnectorConnectionBinding
     * 2.启动LeaderLatchChecker，当zookeeper集群挂掉时{@link Listener#notLeader},唤起检测直到新主节点选出来，并在自己不是主节点时关闭RpcServer，卸载ConnectorConnectionBinding
     * 3.启动RpcClient连接主节点RpcServer
     */
    public void start() {
        try {
            this.leaderLatch.start();
        } catch (Exception e) {
            LOGGER.error("leaderLatch start error: ", e);
        }
        new Thread(this.leaderLatchChecker, "LeaderLatchChecker-thread-" + threadNum.getAndIncrement()).start();
        new Thread(this.rpcClient::start, "RpcClient-thread-" + threadNum.getAndIncrement()).start();
    }

    private void closeRpcIfExist() {
        if (serverThread != null) {
            serverThread.interrupt();

        }
        if (connectorConnectionBinding != null && !connectorConnectionBinding.isClosed()) {
            connectorConnectionBinding.close();
        }

    }

    private void startRpc() {
        if (connectorConnectionBinding != null && connectorConnectionBinding.isClosed()) {
            connectorConnectionBinding.start();
        }

        serverThread = new Thread(this::runRpcServer, "RpcServer-thread-" + threadNum.getAndIncrement());
        LOGGER.info("start rpc server");
        serverThread.start();
    }

    private void runRpcServer() {
        try {
            rpcServer.start();
        } catch (InterruptedException e) {
            LOGGER.error("zookeeper server is error");
        }
    }

    private class Listener implements LeaderLatchListener {

        @Override
        public void isLeader() {
            LOGGER.info("Become leader ,address is: {} !!!", rpcAddress);

            closeRpcIfExist();

            startRpc();
        }

        @Override
        public void notLeader() {
            synchronized (leaderLatchChecker) {
                leaderLatchChecker.notify();
            }
        }
    }

    private class LeaderLatchChecker implements Runnable {

        @Override
        public void run() {
            checkUntilLeaderExist();
        }

        private void checkUntilLeaderExist() {
            boolean retry = false;
            for (; ; ) {
                ThreadUtils.sleep(2000);
                String leaderAddress;
                try {
                    leaderAddress = leaderLatch.getLeader().getId();
                } catch (Throwable t) {
                    LOGGER.error("obtain leaderAddress error: ", t);
                    continue;
                }
                if (leaderAddress == null || leaderAddress.equals("")) {
                    continue;
                }
                synchronized (this) {
                    if (!leaderAddress.equals(rpcAddress)) {
                        closeRpcIfExist();
                    } else {
                        if (retry) {
                            retry = false;
                            continue;
                        }
                    }
                    try {
                        wait();
                        retry = true;
                    } catch (InterruptedException ignore) {
                    }

                }

            }
        }
    }

}
